package com.atguigu.api

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}


/**
 * @description: 从kafka读取后输出到kafka
 * @time: 2020/6/21 13:10
 * @author: baojinlong
 **/
object KafkaSinkTest2 {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

    val inputStream: DataStream[String] = environment.addSource(new FlinkKafkaConsumer011[String]("test1", new SimpleStringSchema(), properties))
    // 基本转换操作
    val outputStream: DataStream[String] = inputStream
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble).toString
      })

    // 查看输入数据
    inputStream.print
    // 写入到kafka:主要还是写入到kafka
    outputStream.addSink(new FlinkKafkaProducer011[String]("localhost:9092", "sinkTestResultTopic", new SimpleStringSchema))

    // 执行
    environment.execute("sink simple test job")

  }
}
